package se.perp.c4.net;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.util.LinkedList;
import java.util.Queue;
import java.util.logging.Level;
import java.util.logging.Logger;

class Feeder {

    private final Queue<Message> incoming,  outgoing;
    private boolean isRunning;
    private final DataInputStream in;
    private final DataOutputStream out;
    private final Object inSync,  outSync;
    private final Logger logger = Logger.getLogger(Feeder.class.getName());
    private final IFeederErrorListener listener;
    private final Runnable sender,  receiver;

    Feeder(DataInputStream istream, DataOutputStream ostream, IFeederErrorListener listener) {
        incoming = new LinkedList<Message>();
        outgoing = new LinkedList<Message>();
        this.in = istream;
        this.out = ostream;
        inSync = new Object();
        outSync = new Object();
        this.listener = listener;
        isRunning = true;
        sender = new Runnable() {

            public void run() {
                try {
                    while (isRunning) {

                        synchronized (outSync) {
                            while (!outgoing.isEmpty()) {
                                Message msg = outgoing.poll();
                                try {
                                    out.writeUTF(msg.getStringData());
                                    if (msg.getType() == Message.Type.PEER_IS_DISCONNECTING) {
                                        isRunning = false;
                                        break;
                                    }
                                } catch (Exception e) {
                                    if (isRunning) {
                                        error(e);
                                        isRunning = false;
                                    }
                                    break;
                                }
                            }
                        }
                        try {
                            Thread.sleep(200);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                } finally {
                    try {
                        out.close();
                    } catch (Exception e) {
                        logger.log(Level.WARNING, "Error while closing output stream.", e);
                    }
                }
            }
        };

        receiver = new Runnable() {

            public void run() {
                try {
                    while (isRunning) {
                        try {
                            String s = in.readUTF();
                            Message m = Message.fromString(s);
                            if (m.getType() == Message.Type.PEER_IS_DISCONNECTING) {
                                isRunning = false;
                                break;
                            } else {
                                synchronized (inSync) {
                                    incoming.add(m);
                                }
                            }
                        } catch (Exception e) {
                            if (isRunning) {
                                error(e);
                                isRunning = false;
                            }
                        }

                        try {
                            Thread.sleep(200);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                } finally {

                    try {
                        in.close();
                    } catch (Exception e) {
                        logger.log(Level.WARNING, "Error while closing input stream.", e);
                    }
                }
            }
        };

        Thread senderThread = new Thread(sender);
        Thread receiverThread = new Thread(receiver);
        senderThread.start();
        receiverThread.start();
    }

    private void error(Exception ex) {
        logger.log(Level.SEVERE, "Connection has been broken!", ex);
        listener.feederError();
        isRunning = false;
    }

    void terminate() {
        queueMessage(Message.newDisconnect());
        try {
            Thread.sleep(300);
        } catch (InterruptedException ie) {
        }
        isRunning = false;
    }

    void queueMessage(Message msg) {

        synchronized (outSync) {
            outgoing.add(msg);
        }
    }

    Message getNextMessage() {
        Message m = null;
        synchronized (inSync) {
            m = incoming.peek();
        }
        return m;
    }

    void removeFirstIncoming() {
        synchronized (inSync) {
            incoming.remove();
        }
    }
}

